package com.coolw.data.migrate.process.strategy;

import com.alibaba.fastjson2.JSON;
import com.coolw.data.migrate.common.alarm.AlarmStatusEnum;
import com.coolw.data.migrate.common.listener.event.BaseEvent;
import com.coolw.data.migrate.common.listener.event.EventTypeEnum;
import com.coolw.data.migrate.common.util.EventUtil;
import com.coolw.data.migrate.common.util.JdbcUtil;
import com.coolw.data.migrate.domain.request.DataMigrateRequest;
import com.coolw.data.migrate.domain.response.DataMigrateResponse;
import lombok.extern.slf4j.Slf4j;
import org.dromara.hutool.core.text.StrUtil;
import org.dromara.hutool.core.text.split.SplitUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.StopWatch;

import java.util.List;

/**
 * 数据迁移处理器
 *
 * @author coolw
 * @date 2024/2/6 11:38
 */
@Slf4j
public abstract class AbstractDataMigrateProcess implements DataMigrateProcess {

    /**
     * 失败批次计数阈值，如果达到阈值，则终止迁移
     */
    private static final int FAIL_BATCH_COUNT_THRESHOLD = 3;

    /**
     * 目标表容量阈值
     */
    @Value("${migrate.targetTableMaxSize}")
    protected int targetTableMaxSize;

    /**
     * 每批次迁移数量
     */
    @Value("${migrate.perBatchLimit}")
    protected int perBatchLimit;

    /**
     * 允许数据迁移的源表名
     */
    @Value("${migrate.sourceTables}")
    protected String sourceTables;

    /**
     * 数据迁移
     * <p>
     * 1.校验源表名是否在配置中,配置了才放行
     * 2.校验目标表是否已达到阈值
     * 3.计算本次数据迁移量
     * 4.按批次数据迁移
     *
     * @param request 数据迁移条件
     * @return 数据迁移结果
     */
    public DataMigrateResponse doProcess(DataMigrateRequest request) {
        String targetTableName = request.getTargetTableName();
        String sourceTableName = request.getSourceTableName();
        String message;

        StopWatch stopWatch = new StopWatch();
        stopWatch.start("数据迁移");

        // 1.校验源表名是否在配置中
        List<String> tables = SplitUtil.split(sourceTables, ",");
        boolean allow = tables.contains(sourceTableName) || tables.contains("*");
        if (!allow) {
            message = StrUtil.format("源表名[{}]未配置,不允许做数据迁移", sourceTableName);
            return fail(request, message, AlarmStatusEnum.NEED_ALARM, stopWatch);
        }

        JdbcUtil.getConnection();
        try {
            // 2.校验目标表是否已达到阈值
            long currentTargetTableSize = JdbcUtil.getTableDataSizeByName(targetTableName);
            if (currentTargetTableSize >= targetTableMaxSize) {
                message = StrUtil.format("目标表[{}]-当前数量[{}],已超过阈值[{}]", targetTableName, currentTargetTableSize, targetTableMaxSize);
                return fail(request, message, AlarmStatusEnum.NEED_ALARM, stopWatch);
            }

            // 3.计算本次数据迁移量
            String getCountSql = getCountSql(request);
            int sourceTotalSize = JdbcUtil.getTableDataSizeBySql(getCountSql);
            log.info("本次源表[{}]向目标表[{}]进行数据迁移,数量:{}", sourceTableName, targetTableName, sourceTotalSize);
            if (sourceTotalSize == 0) {
                message = "无满足源表条件的数据";
                return success(request, 0, 0, message, stopWatch);
            }

            // 4.按批次数据迁移
            return doDataMigrate(request, sourceTotalSize, stopWatch);
        } catch (Exception e) {
            log.error("数据迁移执行发生异常,request:{}", request, e);
            message = "数据迁移执行发生异常";
            return fail(request, message, AlarmStatusEnum.NEED_ALARM, stopWatch);
        } finally {
            JdbcUtil.clear();
        }
    }

    /**
     * 数据迁移
     *
     * @param request         数据迁移条件
     * @param sourceTotalSize 数据迁移量
     * @param stopWatch       计时器
     * @return 数据迁移结果
     */
    private DataMigrateResponse doDataMigrate(DataMigrateRequest request, int sourceTotalSize, StopWatch stopWatch) {
        // 获取数据迁移的SQL
        List<String> sqlList = getDataMigrateSql(request);

        // 当前批次迁移数据量
        int currentMigrateSize = perBatchLimit;
        // 已执行数据迁移量（无论成功与否）
        int migratedSize = 0;
        // 迁移成功数据量
        int successSize = 0;
        // 批次计数
        int batchCount = 0;
        // 失败批次计数
        int failCount = 0;
        String message;
        do {
            // 最后一个批次
            if (migratedSize + perBatchLimit > sourceTotalSize) {
                currentMigrateSize = sourceTotalSize - migratedSize;
                sqlList = getLastDataMigrateSql(request, currentMigrateSize);
            }

            // 校验本次数据迁移量是否小于等于目标表的剩余容量
            message = checkTargetTableThreshold(request.getTargetTableName(), currentMigrateSize);
            if (StrUtil.isNotBlank(message)) {
                if (successSize > 0) {
                    return success(request, sourceTotalSize, successSize, message, stopWatch);
                }
                return fail(request, message, AlarmStatusEnum.NEED_ALARM, stopWatch);
            }

            // 数据迁移SQL执行
            long startTime = System.currentTimeMillis();
            boolean success = JdbcUtil.batchSqlExecuteWithManualTra(sqlList);
            if (success) {
                successSize += currentMigrateSize;
            } else {
                failCount++;
            }
            migratedSize += currentMigrateSize;
            batchCount++;

            long remainSize = sourceTotalSize - migratedSize;
            long cost = System.currentTimeMillis() - startTime;
            log.info("第{}个批次结果:{},迁移数据量:{},剩余:{},耗时:{}ms", batchCount, success, currentMigrateSize, remainSize, cost);

            if (failCount >= FAIL_BATCH_COUNT_THRESHOLD) {
                message = StrUtil.format("数据迁移批次失败次数已达到{}次,终止迁移", FAIL_BATCH_COUNT_THRESHOLD);
                if (successSize > 0) {
                    return success(request, sourceTotalSize, successSize, message, stopWatch);
                }
                return fail(request, message, AlarmStatusEnum.NEED_ALARM, stopWatch);
            }
        } while (sourceTotalSize > migratedSize);

        return success(request, sourceTotalSize, successSize, null, stopWatch);
    }

    /**
     * 校验本次数据迁移量是否小于等于目标表的剩余容量
     *
     * @param targetTableName          目标表名
     * @param currentActualMigrateSize 本次实际迁移数据量
     * @return 检查结果，为null代表通过
     */
    private String checkTargetTableThreshold(String targetTableName, long currentActualMigrateSize) {
        // 当前目标表数据大小
        long currentTargetTableSize = JdbcUtil.getTableDataSizeByName(targetTableName);
        // 检查目标表容量是否上限
        boolean limit = currentTargetTableSize + currentActualMigrateSize > targetTableMaxSize;
        if (limit) {
            return StrUtil.format("目标表[{}]-当前数量[{}],本次迁移数量[{}],已超过阈值[{}]", targetTableName, currentTargetTableSize
                    , currentActualMigrateSize, targetTableMaxSize);
        }
        return null;
    }

    private DataMigrateResponse success(DataMigrateRequest req, int sourceTotalSize, int successSize, String message, StopWatch stopWatch) {
        stopWatch.stop();

        DataMigrateResponse response = DataMigrateResponse.success(message, stopWatch.getTotalTimeMillis());
        response.setSourceTableName(req.getSourceTableName());
        response.setTargetTableName(req.getTargetTableName());
        response.setTotalSize(sourceTotalSize);
        response.setSuccessSize(successSize);
        response.setFailSize(sourceTotalSize - successSize);
        if (sourceTotalSize == successSize) {
            response.setAlarmStatusEnum(AlarmStatusEnum.NO_NEED_ALARM);
        } else {
            response.setAlarmStatusEnum(AlarmStatusEnum.NEED_ALARM);
        }

        // 发送数据迁移成功事件
        EventUtil.publishEvent(new BaseEvent<>(this, EventTypeEnum.DATA_MIGRATE, response, JSON.toJSONString(req)));

        return response;
    }

    private DataMigrateResponse fail(DataMigrateRequest req, String message, AlarmStatusEnum alarmStatusEnum, StopWatch stopWatch) {
        stopWatch.stop();

        DataMigrateResponse response = DataMigrateResponse.fail(message, stopWatch.getTotalTimeMillis());
        response.setSourceTableName(req.getSourceTableName());
        response.setTargetTableName(req.getTargetTableName());
        response.setAlarmStatusEnum(alarmStatusEnum);

        // 发送数据迁移失败事件
        EventUtil.publishEvent(new BaseEvent<>(this, EventTypeEnum.DATA_MIGRATE, response, JSON.toJSONString(req)));

        return response;
    }
}
